﻿using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Logging;
using System.Text;

namespace Common.EventBus.EventBusRabbitMQ
{
    internal class ConsumerRegister : IConsumerRegister
    {
        private readonly ILogger _logger;
        private readonly IServiceProvider _serviceProvider;

        private readonly IConsumerClientFactory _consumerClientFactory;
        private readonly IConsumerServiceSelector _consumerServiceSelector;
        private readonly TimeSpan _pollingDelay = TimeSpan.FromSeconds(1);

        private CancellationTokenSource _cts;
        private Task _compositeTask;
        private bool _disposed;
        private static bool _isHealthy = true;


        public ConsumerRegister(ILogger<ConsumerRegister> logger, IServiceProvider serviceProvider, IConsumerServiceSelector consumerServiceSelector)
        {
            _logger = logger;
            _serviceProvider = serviceProvider;
            _consumerServiceSelector = consumerServiceSelector;
            _consumerClientFactory = serviceProvider.GetService<IConsumerClientFactory>();
            _cts = new CancellationTokenSource();
        }

        public bool IsHealthy()
        {
            return _isHealthy;
        }

        public void Start()
        {
            var dic = _consumerServiceSelector.GetServiceInfo();
            foreach (var (key, value) in dic)
            {
                Task.Factory.StartNew(() =>
                {
                    try
                    {
                        using var client = _consumerClientFactory.Create(key, value);
                        RegisterMessageProcessor(client);
                        client.Subscribe(value);
                        client.Listening(_pollingDelay, _cts.Token);
                    }
                    catch (OperationCanceledException)
                    {
                        //ignore
                    }
                    catch (Exception e)
                    {
                        _logger.LogError(e, e.Message);
                    }
                }, _cts.Token, TaskCreationOptions.LongRunning, TaskScheduler.Default);
            }
            _compositeTask = Task.CompletedTask;
        }

        public void ReStart(bool force = false)
        {
            if (!IsHealthy() || force)
            {
                Pulse();

                _cts = new CancellationTokenSource();
                _isHealthy = true;

                Start();
            }
        }

        public void Dispose()
        {
            if (_disposed)
            {
                return;
            }

            _disposed = true;

            try
            {
                Pulse();

                _compositeTask?.Wait(TimeSpan.FromSeconds(2));
            }
            catch (AggregateException ex)
            {
                var innerEx = ex.InnerExceptions[0];
                if (!(innerEx is OperationCanceledException))
                {
                    _logger.LogError(innerEx, "Dispose error");
                }
            }
        }

        public void Pulse()
        {
            _cts?.Cancel();
        }

        private void RegisterMessageProcessor(IConsumerClient client)
        {
            client.OnMessageReceived += async (sender, queueMessage) =>
            {
                try
                {
                    if (!_consumerServiceSelector.TryGetExecutor(queueMessage.Queue, out var type)) return;
                    using var scope = _serviceProvider.CreateScope();
                    if (scope.ServiceProvider.GetService(type) is IQueueHandler executor)
                    {
                        await executor.Handler(Encoding.UTF8.GetString(queueMessage.Body));
                    }

                    if (scope.ServiceProvider.GetService(type) is IQueueSyncHandler syncExecutor)
                    {
                        syncExecutor.Handler(Encoding.UTF8.GetString(queueMessage.Body));
                    }
                    client.Commit(sender);
                }
                catch (Exception e)
                {
                    _logger.LogError(e, "An exception occurred when process received message. Message:'{0}'.", queueMessage);
                    client.Reject(sender);
                }
            };
        }
    }
}